/**
 * Package name:indi.yangshenhui.tool.task
 * File name:TaskMain.java
 * Date:2016年4月25日-下午5:14:54
 * feiniu.com Inc.Copyright (c) 2013-2015 All Rights Reserved.
 *
 */
package indi.yangshenhui.hbase.dao.task;

import java.io.File;
import java.io.IOException;
import java.security.SecureRandom;
import java.util.Arrays;
import java.util.Random;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hbase.util.Threads;

import indi.yangshenhui.hbase.dao.constant.HBaseConstant;
import indi.yangshenhui.hbase.dao.pool.HTablePoolEngine;

/**
 * @ClassName TaskMain
 * @Description
 * @date 2016年4月25日 下午5:14:54
 * @author shenhui.yang
 * @version 1.0.0
 *
 */
public class TaskMain {
	private static volatile int threadNum = 0;
	private static volatile int targetNum = 0;
	private static volatile boolean verbose = true;
	private static volatile Object verboseLock = new Object();
	private static Random random = new SecureRandom();

	/**
	 * 文件命令。 1)退出应用: echo quit>stop 2)增加10个线程: echo +10>stop 3)减少10个线程: echo
	 * -10>stop 4)设置刚好10个线程: echo 10>stop
	 */
	public static void main(String[] args) throws Exception {
		System.err.println("Usage(V2016-04-25 FOR SHENHUI.YANG): demo [quorum port batchNum threadNum]");
		String quorum = "10.20.16.32,10.20.16.35,10.20.16.36";
		if (args.length > 0) {
			quorum = args[0];
		}
		String port = "2181";
		if (args.length > 1) {
			port = args[1];
		}

		final int batchNum = args.length > 2 ? Integer.valueOf(args[2]) : 1000;

		threadNum = 10;
		if (args.length > 3) {
			threadNum = Integer.valueOf(args[3]);
		}
		targetNum = threadNum;

		HTablePoolEngine.createHBaseConfiguration(HBaseConstant.HBASE_DEFAULT_INSTANCE, quorum, port);
		File stopFile = new File("stop");

		for (int i = 0; i < threadNum; ++i) {
			new Thread(new TaskThread(batchNum)).start();
		}

		FileUtils.write(new File("status"), "Threads:" + threadNum);

		Pattern numCommandPattern = Pattern.compile("[+-]?(\\d+)");

		while (true) {
			while (!stopFile.exists()) {
				Threads.sleep(1000);
			}

			String command = FileUtils.readFileToString(stopFile);
			stopFile.delete();

			command = command.trim();
			syncPrint("Command", command);

			if ("quit".equals(command)) {
				break;
			}

			if (verboseCommand(command)) {
				continue;
			}

			adjustThreadNum(batchNum, numCommandPattern, command);
		}

		synchronized (TaskMain.class) {
			targetNum = 0;
		}

		syncPrint("Main Exited!");
	}

	protected static boolean verboseCommand(String command) {
		if ("echo".equals(command) || "echo on".equals(command)) {
			synchronized (verboseLock) {
				verbose = true;
			}
			return true;
		}

		if ("echo off".equals(command)) {
			synchronized (verboseLock) {
				verbose = false;
			}
			return true;
		}

		return false;
	}

	protected static boolean adjustThreadNum(final int batchNum, Pattern numCommandPattern, String command)
			throws IOException {
		Matcher matcher = numCommandPattern.matcher(command);
		if (!matcher.find()) {
			return false;
		}
		int adjustNum = Integer.parseInt(matcher.group(1));
		synchronized (TaskMain.class) {
			if (matcher.group().startsWith("-"))
				targetNum -= adjustNum;
			else if (matcher.group().startsWith("+"))
				targetNum += adjustNum;
			else
				targetNum = adjustNum;

			if (targetNum < 0)
				targetNum = 0;

			for (int i = 0, ii = targetNum - threadNum; i < ii; ++i) {
				new Thread(new TaskThread(batchNum)).start();
				++threadNum;
			}

			FileUtils.write(new File("status"), "Threads:" + threadNum);
		}

		return true;
	}

	public static class TaskThread implements Runnable {
		private int batchNum;

		public TaskThread(int batchNum) {
			this.batchNum = batchNum;
		}

		@Override
		public void run() {
			syncPrint("Thread starting!");
			Thread.currentThread().setName("HbaseClientThread");
			try {
				batchIntoDb(batchNum);
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				synchronized (TaskMain.class) {
					--threadNum;
					syncPrint("Thread exiting, Left", threadNum);
					try {
						FileUtils.write(new File("status"), "Threads:" + threadNum);
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			}
		}
	}

	protected static void batchIntoDb(int batchNum) {

		long startTime = System.currentTimeMillis();
		long start = 0;
		long c = 0;

		for (long k = 0, kk = 1000L + random.nextInt(10000); !Thread.currentThread().isInterrupted() && k < kk; ++k) {

			if (++c == batchNum) {
				c = 0;

				long end = System.currentTimeMillis();
				syncPrint("Timestamp", "", "Start", start, "Records", batchNum, "End", "", "Cost", end - startTime);

				startTime = end;
			}
		}
	}

	private static void syncPrint(Object... msgParts) {
		synchronized (verboseLock) {
			if (verbose) {
				System.out.println(Arrays.toString(msgParts));
			}
		}
	}

}
